package net.bwie;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsCarBehavior {
    public static void main(String[] args) throws Exception {

        // 创建执行环境
        StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();
        evn.setParallelism(1);

        //  创建表环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(evn);


        // 汇总数据

        // todo 1.车辆进出事实表 ods_cars_log_clean
        tenv.executeSql(
                "CREATE TABLE ods_cars_log_clean (\n" +
                        " id STRING,\n" +
                        " opTime TIMESTAMP(3),\n" +
                        " ctype STRING,\n" +
                        " carCode STRING,\n" +
                        " cId BIGINT,\n" +
                        " proc_time AS PROCTIME(),\n" +
                        // 设置 水位线 3分钟
                        " WATERMARK FOR opTime AS opTime - INTERVAL '3' MINUTES\n" +
                        ") WITH (\n" +
                        " 'connector' = 'kafka',\n" +
                        " 'topic' = 'ods_cars_log_clean',\n" +
                        " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                        " 'properties.group.id' = 'ods_cars_log_clean',\n" +
                        " 'scan.startup.mode' = 'earliest-offset',\n" +
                        " 'format' = 'json'\n" +
                        ")"
        );

        // todo 2.搭建车辆维度表 ods_cars_clean
        tenv.executeSql(
                "CREATE TABLE ods_cars_clean(\n" +
                        "    id INT,\n" +
                        "    owerId INT,\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    type STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH ( \n" +
                        "    'connector' = 'kafka',\n" +
                        "    'topic' = 'ods_cars_clean',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'properties.group.id' = 'ods_cars_clean',\n" +
                        "    'scan.startup.mode' = 'earliest-offset',\n" +
                        "    'format' = 'json'\n" +
                        ")"
        );

        // todo 3.创建 dwd_summary_cars_log 表  车辆进出汇总表
        // 使用窗口TVF方案（Flink 1.13+）
        Table dwd_summary_cars_log = tenv.sqlQuery(
                "SELECT\n" +
                        "  w.id,\n" +  // 使用 w 作为窗口表的别名
                        "  COALESCE(d.owerId, -1) AS owerId,\n" +
                        "  w.opTime,\n" +
                        "  w.cId,\n" +
                        "  w.carCode,\n" +
                        "  w.ctype\n" +
                        "FROM \n" +
                        "TABLE ods_cars_log_clean AS w\n" +
                        "LEFT JOIN ods_cars_clean AS d\n" +
                        "ON w.cId = d.id"
        );

        tenv.createTemporaryView("dwd_summary_cars_log", dwd_summary_cars_log);


//        tenv.sqlQuery("select * from dwd_summary_cars_log limit 10")
//                .execute().print();


//         窗口计算
        tenv.executeSql("CREATE TABLE dws_summary_cars_log (\n" +
                  // 窗口开始时间
                "window_start TIMESTAMP,\n" +
                  // 窗口结束时间
                "window_end TIMESTAMP,\n" +

                // 驶入次数
                "in_cnt INT,\n" +
                // 驶出次数
                "out_cnt INT\n" +

                ") WITH (\n" +
                " 'connector' = 'doris',\n" +
                "'fenodes' = 'node102:9030',\n" +
                "'table.identifier' = 'dws_sca.dws_summary_cars_log',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456',\n" +
                "'sink.label-prefix' = 'doris_cars_log'\n"
        );


        TableResult tableResult = tenv.executeSql("SELECT\n" +

                "window_start,\n"+
                "window_end,\n"+

                "sum(if(ctype = '驶入' ,1,0)) as in_cnt,\n" +
                "sum(if(ctype = '驶出' ,1,0)) as out_cnt\n" +
                 // 设置 滚动窗口
                "FROM (TUMBLE(TABLE dwd_summary_cars_log, DESCRIPTOR(opTime), INTERVAL '3' MINUTES)) as a1  \n" +
                "GROUP BY window_start,window_end;\n"
        );
        tableResult.print();

//        tenv.executeSql("INSERT INTO dws_summary_cars_log\n" +
//                "SELECT\n" +
//                "window_start,\n"+
//                "window_end,\n"+
//
//                "sum(if(ctype = '驶入' ,1,0)) as card_cnt,\n" +
//                "sum(if(ctype = '驶出' ,1,0)) as pw_cnt\n" +
//                // 设置 滚动窗口
//                "FROM (TUMBLE(TABLE dwd_summary_cars_log, DESCRIPTOR(opTime), INTERVAL '3' MINUTES)) a1  \n" +
//                "GROUP BY window_start,window_end;\n"
//        );


        // 触发执行
//        evn.execute("DwsCarBehavior");

    }
}
